package com.atguigu.flink.datastream.transform;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 15:59
 *
 * process算子:
 *    配合处理函数使用。
 *
 *    处理函数的功能:
 *       1. 拥有富函数提供的生命周期方法  以及 可以获取到运行时上下文， 进行状态编程。
 *
 *       2. 可以获取到TimerService ，可以注册定时器(定闹钟) ，计划性的在未来的某个时间做一些处理
 *          当闹钟响了， 会自动调用onTimer方法。 我们可以将处理逻辑写到onTimer方法中。
 *
 *       3. 可以使用侧输出流
 *
 *
 */
public class Flink07_Process {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        ds.process(
                new ProcessFunction<Event, String>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                    }

                    @Override
                    public void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect(JSON.toJSONString(value));
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                    }
                }
        ).print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

